package com.hanxiaozhang.example.listener.other;

import com.hanxiaozhang.constant.RocketConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * 〈一句话功能简述〉<br>
 * 〈基本消息〉
 *
 * @author hanxinghua
 * @create 2022/9/26
 * @since 1.0.0
 */
@Slf4j
@Component
public class No1CommonOriginalSyntax {


    @PostConstruct
    public void init(){

        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketConstant.COMMON_CONSUMER_GROUP);
            consumer.setNamesrvAddr(RocketConstant.NAME_SERVER_ADDR);
            consumer.subscribe(RocketConstant.COMMON_TOPIC, "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExts, ConsumeConcurrentlyContext context) {
                    messageExts.forEach(x->{
                        try {
                            log.info("{}收到消息：{}", this.getClass().getSimpleName(),  new String(x.getBody(), "utf-8"));
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    });
                    // 标记该消息已经被成功消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动消费者实例
            consumer.start();
            log.info("消费者启动成功!");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

}
